package com.study.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * kafka生产者
 * 原理：
 * 1）生产者客户端由两个线程协调运行，主线程和Sender线程
 * 2）主线程：创建消息，通过拦截器、序列化器、分区器之后缓存到累加器(RecordAccumulator)
 * 3）Sender线程：从累加器冲获取消息发送到Kafka
 * 4)累加器作用：缓存消息（默认大小32M），以便Sender线程可以批量发送
 *
 * @author: stephen.shen
 * @create: 2019-06-12 22:08
 */
public class DemoKafkaProducer {

    private final static String brokerList = "localhost:9092";
    private final static String topic = "test-topic";

    private static Properties initConfig() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        // 指定分区中必须要有多少个副本收到这条消息，之后生产者才会认为这条消息是成功写入的
        // 1：生产者发送消息之后，只要分区的 leader 副本成功写入消息，那么它就会收到来自服务端的成功响应
        // 0：生产者发送消息之后不需要等待任何服务端的响应
        // -1 或者 all：生产者在消息发送之后，需要等待 ISR 中的所有副本都成功写入消息之后才能够收到来自服务端的成功响应
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        // 限制生产者客户端能发送的消息的最大值，默认1M
        //props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 1048576);
        // 重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, 1);
        // 重试间隔时间
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100);

        // 数据量达到BATCH_SIZE进行提交
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 100);
        // 如果数据量在LINGER_MS时间内未达到BATCH_SIZE，也会进行提交
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1000);

        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        // 客户端ID
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-1");
        return props;
    }

    public static void main(String[] args) {
        Properties props = initConfig();
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        // 构造消息
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, "kkk", "hello, kafka!");
        // 同步发送
        //producer.send(record).get();
        // 异步发送
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("send success. metadata:" + metadata);
            } else {
                exception.printStackTrace();
            }
        });
        // 关闭
        producer.close();
    }

}